package app.dwd;

import com.bw.gmall.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdDiYiZhiBiao {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        tableEnv.executeSql(MyKafkaUtil.getTopicOPP("order_211126"));
//        tableEnv.sqlQuery("select * from dwd_trade_order_pre_process").execute().print();
        String topic = "dwd_traffic_action_log";
        String groupId = "dwd_traffic_action_log_view_window";
        //总表
        tableEnv.executeSql("create table action_log( " +
                "    `common` map<string,string>, " +
                "    `action` map<string,string>, " +
                "    `page` map<string,string>, " +
                "    `ts` bigint, " +
                "    `rt` as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)), " +
                "    WATERMARK FOR rt AS rt - INTERVAL '2' SECOND " +
                " ) " + MyKafkaUtil.getKafkaDDL(topic, groupId));

        Table table1 = tableEnv.sqlQuery("select " +
                "CONCAT('shop_', CAST(CAST(FLOOR(RAND() * 10) + 1 AS INT) AS STRING)) AS shop_id, " +
                "`common`['uid'] as uid, " +
                "`page`['item'] as sku_id," +
                "`ts`" +
                "from action_log " +
                "where `action`['action_id'] = 'cart_add'");

        tableEnv.createTemporaryView("cart_log",table1);

        tableEnv.executeSql("create table dwd_cart_add_count(" +
                "shop_id string, " +
                "uid string, " +
                "sku_id string," +
                "ts BIGINT," +
                "primary key(shop_id) not enforced" +
                ")"+ MyKafkaUtil.getUpsertKafkaDDL("dwd_cart_add_count"));

        tableEnv.executeSql("insert into dwd_cart_add_count select * from cart_log");

    }
}
